package com.ehotting.mqbd.core.call;

import com.ehotting.eaf.core.DateHelper;
import com.ehotting.eaf.core.snow.SnowflakeUtil;
import com.ehotting.eaf.core.utils.AssertUtils;
import com.ehotting.eaf.core.utils.StringUtils;
import com.ehotting.eaf.core.utils.ip.IpUtils;
import com.ehotting.mqbd.api.MQDBConstants;
import com.ehotting.mqbd.api.MqDataStateConstants;
import com.ehotting.mqbd.api.TaskStateActionConstants;
import com.ehotting.mqbd.api.bean.*;
import com.ehotting.mqbd.api.bean.solver.SolverStateBean;
import com.ehotting.mqbd.api.enums.DirectionEnum;
import com.ehotting.mqbd.api.enums.MsgStateEnum;
import com.ehotting.mqbd.api.enums.ProtocolTypeEnum;
import com.ehotting.mqbd.api.enums.TopicEnum;
import com.ehotting.mqbd.api.exception.CallException;
import com.ehotting.mqbd.api.exception.TcpException;
import com.ehotting.mqbd.core.service.MQService;
import com.ehotting.mqbd.core.logger.LOGGER_KEY;
import com.ehotting.mqbd.core.manager.*;
import com.ehotting.mqbd.core.service.TaskStateMQService;
import lombok.SneakyThrows;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.Date;
import java.util.List;

public abstract class CallerAbstract implements CallerAble {

    private Logger logger = LoggerFactory.getLogger(CallerAbstract.class);

    private MQService mqService;

    protected SolverStateManager solverStateManager;
    @Autowired
    protected MsgStateManager msgStateManager;
    @Autowired
    protected MqQueueManager mqQueueManager;
    @Autowired
    protected AgentLinkManager agentLinkManager;
    @Autowired
    protected TaskStateMQService taskStateMQService;

    @Override
    public void invoke(SolverStateBean solverState, MQDataAble data) throws Exception {
        /**数据验收**/
        validData(data);

        if(data.isSync()){
            execute(solverState,data);
        }else{
            Runnable run = new Runnable() {
                @SneakyThrows
                @Override
                public void run() {
                    try {
                        addMDC(data);
                        execute(solverState, data);
                    }catch (Exception e){
                        logger.error("Runnable.run()",e);
                    }finally {
                        MDC.clear();
                    }
                }
            };
            Thread t = new Thread(run);
            t.start();
        }
    }

    private void execute(SolverStateBean solverState, MQDataAble mqData) throws Exception {
        boolean isSuccess = true;
        try {
            AssertUtils.notNull(solverState,"Module["+mqData.getModule()+"] solverState is null or offline,please check.");
            broadcastQueueMsg(mqData,solverState);
            beforeExecute(mqData);
            doTaskStart(mqData);
            doAction(solverState,mqData);
            doTaskFinish(mqData);
        } catch (Exception e) {
            if(e instanceof TcpException){
                TcpException ee = (TcpException)e;
                mqData.setState(ee.getCode());
                mqData.setMsg(ee.getMsg());
            }else if(e instanceof CallException){
                CallException ee = (CallException)e;
                mqData.setState(ee.getCode());
                mqData.setMsg(ee.getMessage());
            } else {
                mqData.setState(MqDataStateConstants.SYS_ERROR);
                mqData.setMsg(e.getMessage());
            }
            isSuccess = false;
            logger.error("execute",e);
            throw e;
        }finally {
            response(mqData);
            if(isSuccess == false){
                doTaskFail(mqData);
            }
            afterExecute( solverState, mqData);
        }
    }

    private void addMDC(MQDataAble data){
        MDC.put(LOGGER_KEY.TRACE_ID, SnowflakeUtil.nextId64());
        MDC.put(LOGGER_KEY.MODULE,data.getModule());
    }

    private void beforeExecute(MQDataAble baseData){
        TopicEnum topicEnum = baseData.getTopic();
        if(topicEnum == TopicEnum.BLOCK){
            msgStateManager.setSolving(baseData.getMsgId());
        }
    }

    private void afterExecute(SolverStateBean solverState, MQDataAble data){

        if(data.getTopic() == TopicEnum.BLOCK && data.getProtocolType() == ProtocolTypeEnum.HTTP){
            solverStateManager.setSolverFree(solverState);
            msgStateManager.deleteByMsgId(data.getMsgId());
        }

        if(data.getProtocolType() == ProtocolTypeEnum.HTTP){
            MDC.clear();
        }
    }

    private void response(MQDataAble data){
        setReturnParam(data);
        mqService.request(data);
    }

    private void setReturnParam(MQDataAble data){
        data.setTopic(TopicEnum.CONCURRENCY);
        data.setDirection(DirectionEnum.RESPONSE.getCode());
        String tempHost = data.getLocalHost();
        data.setLocalHost(data.getConsumerHost());
        data.setConsumerHost(tempHost);
    }


    public String replaceUrl(SolverStateBean solverState, String url){
        if(StringUtils.isEmpty(url)){
            return null;
        }
        logger.info("oldUrl:{}",url);
        logger.info("solverState.getSolverInfo().toStringIpAndPort():{}",solverState.getSolverInfo().toStringIpAndPort());
        String replacement = solverState.getSolverInfo().toStringIpAndPort();
        String newUrl = IpUtils.replaceIpPort(url,replacement);
        logger.info("newUrl:{}",newUrl);
        return newUrl;
    }

    public void setMqService(MQService mqService) {
        this.mqService = mqService;
    }

    public void setSolverStateManager(SolverStateManager solverStateManager) {
        this.solverStateManager = solverStateManager;
    }



    public void broadcastQueueMsg(MQDataAble mqData,SolverStateBean solverState){
        logger.info("Start broadcastQueueMsg.");

        if(mqData.getTopic()== TopicEnum.CONCURRENCY){
            logger.info("Broadcast queue by mqData is  Concurrency,return.");
            return;
        }

        MsgStateBean currentMsgState = msgStateManager.getByMsgId(mqData.getMsgId());

        AssertUtils.notNull(currentMsgState,"currentMsgState[module:"+mqData.getModule()+",agentId:"+mqData.getAgentId()+",msgId:"+mqData.getMsgId()+"] is null,return.");

        msgStateManager.clearInvalidCancelAndQueueState(mqData.getModule(),currentMsgState);

        List<MsgStateBean> msgStateList = msgStateManager.findInQueueByModuleAndInQueueAndSorted(mqData.getModule(),currentMsgState.getDate());

        logger.info("msgStateList.size():{}",msgStateList.size());

        if(msgStateList.size()==0){
            logger.error("msgStateList.size() is 0,return.");
            return;
        }

        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }


        QueueNotifyAgentAble notifyAgent = new QueueNotifyAgentAble(
                mqQueueManager,
                msgStateManager,
                agentLinkManager,
                mqService){
            @Override
            public MQDataAble genQueueMsg(long queueIndex,MsgStateBean msgState,SolverStateBean solverState) {
                MQDataAble data = genMqQueueData(mqData.getModule(),queueIndex,msgState,solverState);
                return data;
            }
        };



        Runnable broadcastRun = new Runnable() {
            @Override
            public void run() {
                notifyAgent.invoke(mqData,msgStateList,solverState);
            }
        };
        Thread t = new Thread(broadcastRun);
        t.start();

        logger.info("Start broadcastQueueMsg.");
    }

    public MQDataAble genMqQueueData(String module,long queueIndex,MsgStateBean msgState,SolverStateBean solverState){

        MQHttpData data = MQHttpData.create(TopicEnum.CONCURRENCY,
                null,
                MQDBConstants.DISPATCHER_HOST,
                msgState.getAgentId(),
                module,"");

        Long userId = agentLinkManager.getUserId(msgState.getAgentId());
        Long mainProjectId = agentLinkManager.getProjectId(msgState.getAgentId());

        String projectNo = "";

        String moduleName = getModuleName();


        String time = "["+ DateHelper.toSeparatorStringyyyyMMddHHmmss(new Date())+"]";
        if(mainProjectId>0){
            projectNo = "[编号:"+mainProjectId+"]";
        }


        String msg = time + moduleName + projectNo+"排队为第 "+queueIndex+" 位";

        if(queueIndex <= 0){
            msg = time + moduleName + projectNo+"排队已到，求解器为："+solverState.getSolverInfo().getIp()+"，开始求解";
        }


        if(msgState!=null && msgState.getState()== MsgStateEnum.SOLVING){
            msg = time+ moduleName +projectNo+"，求解器为："+solverState.getSolverInfo().getIp()+",正在求解中";
        }

        logger.info("msg:{}.",msg);

        CommonResponse response = new CommonResponse();
        response.setUserId(userId);
        response.setMsg(msg);
        response.setId(String.valueOf(mainProjectId));
        response.setAction(TaskStateActionConstants.TASK_QUEUE);

        data.setResponse(response);

        return data;
    }

    public void doTaskStart(MQDataAble mqData){
        if(mqData.getTopic() == TopicEnum.CONCURRENCY){
            logger.info("doTaskStart is  Concurrency,return.");
            return;
        }

        MsgStateBean currentMsgState = msgStateManager.getByMsgId(mqData.getMsgId());

        AssertUtils.notNull(currentMsgState,"currentMsgState[module:"+mqData.getModule()+",agentId:"+mqData.getAgentId()+",msgId:"+mqData.getMsgId()+"] is null,return.");

        if(currentMsgState.getState() == MsgStateEnum.CANCEL){
            taskStateMQService.taskCancel(this.mqService,mqData.getAgentId());
        }else{
            taskStateMQService.taskStart(this.mqService,mqData.getAgentId());
        }

    }


    public void doTaskFinish(MQDataAble mqData){
        if(mqData.getTopic()== TopicEnum.CONCURRENCY){
            logger.info("doTaskFinish is  Concurrency,return.");
            return;
        }
        if(mqData.getProtocolType() != ProtocolTypeEnum.HTTP ){
            logger.info("doTaskFinish is not  http,return.");
            return;
        }
        taskStateMQService.taskFinish(this.mqService,mqData.getAgentId());
    }


    public void doTaskFail(MQDataAble mqData){
        if(mqData.getTopic()== TopicEnum.CONCURRENCY){
            logger.info("doTaskFinish is  Concurrency,return.");
            return;
        }

        taskStateMQService.taskFail(this.mqService,mqData.getAgentId());

        msgStateManager.deleteByMsgId(mqData.getMsgId());
    }

    public abstract String getModuleName();
}
